package com.zhou.publish.confirm;

/**
 * NewTask
 *
 * @author zhouxianfeng
 * @date 2021-5-24
 */

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

@Slf4j
@Component
public class NewTask {
    private final static String QUEUE_NAME = "test_queue_confirm1";

    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("10.101.75.152");
        factory.setUsername("user2");
        factory.setPassword("user2");

        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            channel.confirmSelect();

            String message = "hello confirm message";
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            if (!channel.waitForConfirms()) {
                System.out.println("message send failed");
            } else {
                System.out.println("message send ok");
            }
        }
    }

    @Autowired
    private RabbitTemplate template;

    @Autowired
    private Queue confirmQueue;

    //@Scheduled(fixedDelay = 1000, initialDelay = 500)
    public void test() throws InterruptedException, ExecutionException, TimeoutException {
        CorrelationData cd1 = new CorrelationData();
        this.template.convertAndSend("", confirmQueue.getName(), "foo", cd1);
        CorrelationData.Confirm confirm = cd1.getFuture().get(1000, TimeUnit.SECONDS);
        if (!confirm.isAck()) {
            log.info("发送失败");
        }
        log.info("{}", confirm);
    }

    @RabbitListener(queues = "#{confirmQueue.name}")
    public void receive(String in) {
        System.out.println(" [x] Received '" + in + "'");
    }

/*    public void test1() {
        Collection<?> messages = null;
        Boolean result = this.template.invoke(t -> {
            messages.forEach(m -> t.convertAndSend(confirmQueue.getName(), m));
            t.waitForConfirmsOrDie(10_000);
            return true;
        }, (tag, multiple) -> {
            log.info("Ack: " + tag + ":" + multiple);
        }, (tag, multiple) -> {
            log.info("Nack: " + tag + ":" + multiple);
        }));
    }*/

}
